#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdint.h>

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <stdint.h>

#include <time.h>
#include <ctype.h>
#include <stddef.h>
#include <stdio.h>
#include <stdint.h>
#include "os_api.h"
#include "ps_api.h"
#include "dev_api.h"
#include "debug.h"
#include "sockets.h"
#include "def.h"
#include "netdb.h"
#include "socket_service.h"


#define _TASK_STACK_SIZE    (1024*4)
#define SOCKET_CALLBACK(evt, param); do {  \
										{ \
											if(g_s_sockets[app_id].callback) \
											{ \
												g_s_sockets[app_id].callback(app_id, evt, param); \
											}\
										} \
									} while(0) 

#define USER_SOCKET_MAX_COUNT 6
//socket数据发送task mailbox长度
#define SEND_TASK_MAILBOX_EVENT_MAX_SIZE 50
#define SOCKET_TASK_NAME_MAX_LEN 20
#define SOCKET_MSG_NAME_MAX_LEN 20 

typedef struct socket_item_info
{
	int app_id;
	int socket_id;
	int protocol; //0:tcp 1:udp
	char *address;
	int port;
	int is_connected;
	int connect_task_exist;
	socket_service_event_callback_t callback;
	char connect_task_name[SOCKET_TASK_NAME_MAX_LEN+1];
	OSTaskRef connect_task_handle;
    void *connect_task_stack;
	char send_task_name[SOCKET_TASK_NAME_MAX_LEN+1];
	OSTaskRef send_task_handle;
    OSMsgQRef send_task_msg;
	char send_msg_name[SOCKET_MSG_NAME_MAX_LEN+1];
    void *send_task_stack;

	char recv_task_name[SOCKET_TASK_NAME_MAX_LEN+1];
	OSTaskRef recv_task_handle;
    void *recv_task_stack;

	OSSemaRef connect_ok_semaphore;
	UINT32 recv_semaphore;
    int connect_flag;
    
}socket_item_info_t;

typedef struct socket_send_data
{
    char *data;
    uint32_t len;
    int user_param;
}socket_send_data_t;


static socket_item_info_t g_s_sockets[USER_SOCKET_MAX_COUNT];

static int socket_exist(int id) 
{
    return (id >= 0 && id < USER_SOCKET_MAX_COUNT) ? 1 : 0;
}

static void close_socket(int app_id)
{
	LOG_PRINTF("enter close_socket");
	close(g_s_sockets[app_id].socket_id);
	g_s_sockets[app_id].socket_id = -1;
	g_s_sockets[app_id].is_connected = 0;
}

static void reconnect_socket(int app_id)
{
	LOG_PRINTF("reconnect_socket app_id = %d",app_id);
	socket_connect(app_id, 
		g_s_sockets[app_id].protocol, 
		g_s_sockets[app_id].address, 
		g_s_sockets[app_id].port, 
		g_s_sockets[app_id].callback);
}


static void send_task_proc(void *arg)
{

	int ret = -1;
	uint32_t sent_len = 0;
	int app_id = *(int*)(arg);
	socket_event_param_t event_param;
	socket_send_data_t data_item = {0};
	while(1)
	{
		OSAMsgQRecv(g_s_sockets[app_id].send_task_msg, (UINT8 *)&data_item, (UINT32)sizeof(socket_send_data_t), OS_SUSPEND);
        if(g_s_sockets[app_id].is_connected)
        {
            sent_len = 0;
            LOG_PRINTF("total len %d, sent len %d", data_item.len, sent_len);
            while (sent_len < data_item.len)
            {
                ret = send(g_s_sockets[app_id].socket_id, data_item.data+sent_len, data_item.len-sent_len, 0);
                if (ret >= 0)
                {
                    LOG_PRINTF("send %d bytes", ret);
                    sent_len += ret;
                    if (sent_len >= data_item.len)
                    {
                        event_param.send_cnf_t.result = sent_len;
                        event_param.send_cnf_t.user_param = data_item.user_param;
                        SOCKET_CALLBACK(SOCKET_EVENT_SEND, event_param);
                    }						
                }
                else
                {
                    if (lwip_getsockerrno(g_s_sockets[app_id].socket_id) == EWOULDBLOCK)
                    {
                        LOG_PRINTF("block, wait send buffer ok");
                        OSATaskSleep(200);
                    }
                    else
                    {
                        close_socket(app_id);

                        event_param.send_cnf_t.result = ret;
                        event_param.send_cnf_t.user_param = data_item.user_param;
                        SOCKET_CALLBACK(SOCKET_EVENT_SEND, event_param);
						free(data_item.data);
       					data_item.data = NULL;
                        OSATaskSleep(200 * 5);
						
                        reconnect_socket(app_id);

                        break;
                    }
                }					
            }				
        }
        else
        {
            ret = -1;
            event_param.send_cnf_t.result = ret;
            event_param.send_cnf_t.user_param = data_item.user_param;
            SOCKET_CALLBACK(SOCKET_EVENT_SEND, event_param);
        }

        free(data_item.data);
        data_item.data = NULL;
    }
}


#define RECV_BUF_LEN 1024


static void recv_task_proc(void *arg)
{	
	fd_set read_set, error_set;
	struct timeval timeout;
	int ret, read_len;
	char * recv_buf = NULL;
	int app_id = *(int*)(arg);
	socket_event_param_t event_param;

	LOG_PRINTF("enter recv_task_proc");

	while(1)
	{
		if(g_s_sockets[app_id].is_connected)
		{						
			OSASemaphoreAcquire(g_s_sockets[app_id].connect_ok_semaphore, OS_NO_SUSPEND);

			if(NULL == recv_buf)
			{
				recv_buf = (char *)malloc(RECV_BUF_LEN);
				LOG_PRINTF("malloc recv_buf fail");
			}

			timeout.tv_sec = 60;
			timeout.tv_usec = 0;

			while(1)
			{
				FD_ZERO(&read_set);
				FD_ZERO(&error_set);
				FD_SET(g_s_sockets[app_id].socket_id,&read_set);
				FD_SET(g_s_sockets[app_id].socket_id,&error_set);

				LOG_PRINTF("before select");

				ret = select(g_s_sockets[app_id].socket_id+1, &read_set, NULL, &error_set, &timeout);

				if(ret < 0)
				{
					//失败
					LOG_PRINTF("select fail, ret %d",ret);
					break;
				}
				else if(ret == 0)
				{
					//超时
					LOG_PRINTF("select timeout");
				}
				else
				{
					if(FD_ISSET(g_s_sockets[app_id].socket_id, &error_set))
					{
						//出错
						LOG_PRINTF("select error event");
						break;
					}
					else if(FD_ISSET(g_s_sockets[app_id].socket_id, &read_set))
					{
						LOG_PRINTF("select read event");
						read_len = recv(g_s_sockets[app_id].socket_id, recv_buf, RECV_BUF_LEN, 0);
						//read event后，第一次读不到数据，表示出错
						//实测，在select时，服务器主动断开连接，会走到这里
						if(read_len <= 0)
						{
							LOG_PRINTF("select read event error");
							break;
						}
						else
						{
							do
							{
								LOG_PRINTF("recv %d bytes data from server",read_len);
								if(read_len > 0)
								{
									//读到了数据，在recv_buf中，长度为read_len
									event_param.recv_ind_t.len = read_len;
									event_param.recv_ind_t.data = recv_buf;
									SOCKET_CALLBACK(SOCKET_EVENT_RECEIVE, event_param);
								}

								read_len = recv(g_s_sockets[app_id].socket_id, recv_buf, RECV_BUF_LEN, 0);
							}while(read_len > 0);
						}					

					}
					else
					{
						LOG_PRINTF("select other socket event");
					}
				}
			}

			close_socket(app_id);			

			OSATaskSleep(200 * 5);
			reconnect_socket(app_id);					
		}
		else
		{
			//等待connect ok
			LOG_PRINTF("wait connect ok semaphore");
			OSASemaphoreAcquire(g_s_sockets[app_id].connect_ok_semaphore, 200);
		}
	}
	
}


static void connect_task_proc(void *arg)
{
	int ret;
	ip_addr_t remote_ip;
    struct sockaddr_in name;
    socklen_t sockaddr_t_size = sizeof(name);
    int	h_errnop;
    struct hostent dns_result;
    struct hostent *p_result;
	int app_id = *(int*)(arg);
	socket_event_param_t event_param;

	LOG_PRINTF("Enter %s",__FUNCTION__);
	while(1)
    {
		//检查网络是否准备就绪，如果尚未就绪，等待1秒后循环重试
		while(!ps_GetNetworkReady())
		{
			LOG_PRINTF("wait network_service_is_ready");
			OSATaskSleep(200);
		}

		//执行DNS，如果失败，等待1秒后，返回检查网络逻辑，重试
		char buf[128] = {0};
		dns_result = *(gethostbyname(g_s_sockets[app_id].address));
		if(dns_result.h_length != 0)
		{
			remote_ip = *((ip_addr_t *)dns_result.h_addr_list[0]);
		}
		else
		{
			OSATaskSleep(200);
			LOG_PRINTF("dns fail");
			continue;
		}
		//创建套接字，如果创建失败，等待3秒后循环重试
		int socket_id = socket(AF_INET, 
							   (g_s_sockets[app_id].protocol)==1 ? SOCK_DGRAM : SOCK_STREAM, 
							   (g_s_sockets[app_id].protocol)==1 ? IPPROTO_UDP : IPPROTO_TCP);
		if(socket_id < 0)
		{
			LOG_PRINTF("create socket fail");
			OSATaskSleep(200 * 3);
			continue;
		}
		g_s_sockets[app_id].socket_id = socket_id;
		
		//连接服务器，如果失败，关闭套接字，等待5秒后，返回检查网络逻辑，重试
		name.sin_family = AF_INET;
		name.sin_addr.s_addr = remote_ip.addr;
		name.sin_port = htons(g_s_sockets[app_id].port);

        ret = connect(socket_id, (const struct sockaddr *)&name, sockaddr_t_size);
		if(ret < 0)
		{
			LOG_PRINTF("connect fail, ret %d",ret);
			close_socket(app_id);
			event_param.connect_result = ret;
			SOCKET_CALLBACK(SOCKET_EVENT_CONNECT, event_param);			
			OSATaskSleep(200 * 5);
			continue;
		}
		LOG_PRINTF("connect ok");
		g_s_sockets[app_id].is_connected = 1;
		
		fcntl(socket_id, F_SETFL, O_NONBLOCK);

		if (NULL == g_s_sockets[app_id].connect_ok_semaphore)
		{
			OSASemaphoreCreate (&g_s_sockets[app_id].connect_ok_semaphore, 1, OSA_FIFO);			
		}
		else
		{
			OSASemaphoreRelease(g_s_sockets[app_id].connect_ok_semaphore);		   
		}		

		if(g_s_sockets[app_id].send_task_handle == NULL)
		{
			snprintf(g_s_sockets[app_id].send_task_name, SOCKET_TASK_NAME_MAX_LEN, "%s_%d", "socket_send", app_id);
			g_s_sockets[app_id].send_task_stack = (void *)malloc(_TASK_STACK_SIZE);
			OSATaskCreate(&g_s_sockets[app_id].send_task_handle,g_s_sockets[app_id].send_task_stack,_TASK_STACK_SIZE,80,g_s_sockets[app_id].send_task_name,send_task_proc,&g_s_sockets[app_id].app_id);
		}
           
		if(g_s_sockets[app_id].recv_task_handle == NULL)
		{
			snprintf(g_s_sockets[app_id].recv_task_name, SOCKET_TASK_NAME_MAX_LEN, "%s_%d", "socket_recv", app_id);
			g_s_sockets[app_id].recv_task_stack = (void *)malloc(_TASK_STACK_SIZE);
            OSATaskCreate(&g_s_sockets[app_id].recv_task_handle,g_s_sockets[app_id].recv_task_stack,_TASK_STACK_SIZE,80,g_s_sockets[app_id].recv_task_name,recv_task_proc,&g_s_sockets[app_id].app_id);
		}

		event_param.connect_result = ret;
		SOCKET_CALLBACK(SOCKET_EVENT_CONNECT, event_param);	

		break;				
    }

	LOG_PRINTF("exit");
	g_s_sockets[app_id].connect_task_exist = 0;
	if(g_s_sockets[app_id].connect_task_handle != NULL){
		OSATaskDelete(g_s_sockets[app_id].connect_task_handle);
	}
}

int socket_connect(int app_id, int protocol, char *address, int port, socket_service_event_callback_t callback)
{
	LOG_PRINTF("Enter %s",__FUNCTION__);
	if (!socket_exist(app_id))
	{
		return -1;
	}
	
	if(!g_s_sockets[app_id].connect_task_exist)
	{
		g_s_sockets[app_id].app_id = app_id;
		g_s_sockets[app_id].protocol = protocol;

		if (!g_s_sockets[app_id].address)
		{
			free(g_s_sockets[app_id].address);
			g_s_sockets[app_id].address = NULL;
		}
		g_s_sockets[app_id].address = (char *)malloc(strlen(address)+1);
		memset(g_s_sockets[app_id].address, 0, strlen(address)+1);
		memcpy(g_s_sockets[app_id].address, address, strlen(address));
		
		g_s_sockets[app_id].port = port;
		g_s_sockets[app_id].callback = callback;
		
		if(g_s_sockets[app_id].connect_task_name != NULL){
			memset(g_s_sockets[app_id].connect_task_name,0x00,SOCKET_TASK_NAME_MAX_LEN);
		}
		snprintf(g_s_sockets[app_id].connect_task_name, SOCKET_TASK_NAME_MAX_LEN, "%s_%d", "socket_connect", app_id);
		if(g_s_sockets[app_id].send_msg_name != NULL){
			memset(g_s_sockets[app_id].send_msg_name,0x00,SOCKET_MSG_NAME_MAX_LEN);
		}
		snprintf(g_s_sockets[app_id].send_msg_name, SOCKET_MSG_NAME_MAX_LEN, "%s_%d", "socket_send", app_id);
		if(g_s_sockets[app_id].send_task_msg == NULL){
			OSAMsgQCreate(&g_s_sockets[app_id].send_task_msg,g_s_sockets[app_id].send_msg_name,(UINT32)sizeof(struct socket_send_data),10,OS_FIFO);
		}
		g_s_sockets[app_id].connect_task_exist = (OSATaskCreate(&g_s_sockets[app_id].connect_task_handle,g_s_sockets[app_id].connect_task_stack,_TASK_STACK_SIZE,80,g_s_sockets[app_id].connect_task_name,connect_task_proc,&g_s_sockets[app_id].app_id)==OS_SUCCESS);
	}

	return 	g_s_sockets[app_id].connect_task_exist ? 0 : -1;
}

//同步插入队列结果通过返回值判断：0成功，其他失败
//插入成功后，异步发送结果通过callback通知
int socket_send(int app_id, uint32_t len, const char *data, int user_param)
{
	LOG_PRINTF("enter socket_send");
	if (!socket_exist(app_id))
	{
		return -1;
	}

	if(data==NULL || len==0)
	{
		return -1;
	}
	socket_send_data_t data_item = {0};

	data_item.data = (char *)malloc(len);
	if(data_item.data == NULL){
		LOG_PRINTF("malloc data_item.data fail");
		ASSERT(OS_FAIL);
	}
	memcpy(data_item.data, data, len);
	data_item.len = len;
	data_item.user_param = user_param;

	int ret = OSAMsgQSend(g_s_sockets[app_id].send_task_msg,(UINT32)sizeof(struct socket_send_data),(UINT8 *)&data_item,OSA_NO_SUSPEND);

	LOG_PRINTF("end socket_send");
	return ret;
}
